package org.example.flinksql.operation;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.row;

public class Filter {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // create an input table
        Table A = tbEnv.fromValues(
                DataTypes.ROW(
                    DataTypes.FIELD("id", DataTypes.BIGINT()),
                    DataTypes.FIELD("name", DataTypes.STRING()),
                    DataTypes.FIELD("age", DataTypes.INT()),
                    DataTypes.FIELD("sex", DataTypes.STRING())
                ),
                row(1, "user-1", 22, "男"),
                row(2, "user-2", 26, "男"),
                row(2, "user-3", 22, "女"),
                row(4, "user-4", 23, "男"),
                row(5, "user-5", 21, "女"),
                row(6, "user-6", 21, "女"),
                row(7, "user-7", 21, "女"),
                row(8, "user-8", 21, "男"),
                row(9, "user-9", 21, "女")
        );

        // create an output table
        Table B = tbEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("id", DataTypes.BIGINT()),
                        DataTypes.FIELD("name", DataTypes.STRING()),
                        DataTypes.FIELD("age", DataTypes.INT())
                ),
                row(0, "user-0", 22),
                row(4, "user-4", 23)
        );

        /* print table struct
        root
                |-- id: BIGINT
                |-- name: STRING
                |-- age: INT
                |-- sex: STRING

        root
                |-- id: BIGINT
                |-- name: STRING
                |-- age: INT
        */
        A.printSchema();
        B.printSchema();

        // select all records from A
        TableResult resultA = A.select($("*")).execute();
        resultA.print();
        // select all records from B
        TableResult resultB = B.select($("*")).execute();
        resultB.print();

        // filter and limit records
        TableResult result1 = A.filter($("sex").isEqual("男"))
                .select($("*"))
                .limit(3)
                .execute();
        result1.print();




        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
